package server

import (
	"context"
	"gitee.com/tjccc/gomq/mq/broker"
	"net"
	"sync"
	"sync/atomic"

	"gitee.com/tjccc/gomq/lib/logger"
	"gitee.com/tjccc/gomq/mq/connection"
	parser "gitee.com/tjccc/gomq/parser/custom"
)

// 根据topic 放到对应队列
type Handler struct {
	activeConn sync.Map
	closing    atomic.Bool
	broker     *broker.BrokerEngine
}

func MakeHandler() *Handler {
	engine := broker.NewBrokerEngine()
	return &Handler{
		broker: engine,
	}
}

func (h *Handler) Handle(ctx context.Context, conn net.Conn) {
	if h.closing.Load() {
		_ = conn.Close()
		return
	}
	client := connection.NewConn(conn)
	ch := parser.ParseStream(conn)
	for payload := range ch {
		if payload.Err != nil {
			logger.Errorf("receive payload error:" + payload.Err.Error())
			_ = conn.Close()
			break
		}
		reply := h.broker.Exec(client, payload.Data)
		if reply != nil {
			_, err := conn.Write(reply.ToBytes())
			if err != nil {
				logger.Errorf("connection closed:%d", client.GetClientId())
				_ = conn.Close()
				break
			}
		}
	}
}

func (h *Handler) Close() error {
	h.closing.Store(true)
	h.activeConn.Range(func(key interface{}, val interface{}) bool {
		client := key.(*connection.Connection)
		_ = client.Close()
		return true
	})
	return nil
}
